package com.wei.outputformat;

import java.io.IOException;
import java.util.HashMap;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.wei.flow.FlowBean;
import com.wei.flow.FlowJob;
import com.wei.flow.FlowMap;
import com.wei.flow.FlowPartitioner;
import com.wei.flow.FlowReduce;

public class LogEnhancer {
	
	static class LogEnhancerMap extends Mapper<LongWritable, Text, Text, NullWritable>{
		HashMap<String, String> ruleMap=new HashMap<>();
		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			DBLoader.dbLoader(ruleMap);
		}
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] fields = value.toString().split("\t");
			try {
				String string = fields[26];
				String string2 = ruleMap.get(string);
				String result="";
				if (StringUtils.isNotBlank(string2)) {
					result=value.toString()+"\t"+string2+"\n";
				}else{
					result=value.toString()+"toCrawl\n";
				}
				context.write(new Text(result), NullWritable.get());
			} catch (Exception e) {
				
			}
		}

	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(LogEnhancer.class);
		job.setMapperClass(LogEnhancerMap.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		job.setOutputFormatClass(LogEnhancerOutputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}

}
